package com.ztrue.framework.core.event.impl;

import com.mongodb.MongoClient;
import com.mongodb.MongoClientOptions;
import com.mongodb.ServerAddress;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.MongoDatabase;
import com.ztrue.framework.core.event.DomainEventStream;
import com.ztrue.framework.core.event.IDomainEvent;
import com.ztrue.framework.core.event.IEventSerializer;
import com.ztrue.framework.core.event.IEventStore;
import com.ztrue.framework.infrastructure.impl.DefaultClassProvider;
import com.ztrue.common.log.Logger;
import com.ztrue.common.properties.Proper;
import com.ztrue.common.serializing.ITextSerializer;
import com.ztrue.common.serializing.impl.FastjsonSerializer;
import com.ztrue.common.utilities.Assert;
import org.bson.Document;

import static com.mongodb.client.model.Filters.*;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * 表示一个使用MongoDB作为事件存储数据库
 *
 * @author zhangw
 * @date 2016/11/23 16:56
 * @version: 1.0
 */
public class MongoDBEventStore implements IEventStore {
    private static final Logger logger = Logger.getLogger(MongoDBEventStore.class);
    private final MongoClient mongoClient;
    private final MongoDatabase db;
    private final String DatabaseName;
    private final String COLLECTIONNAME = "events";
    private final ITextSerializer serializer;
    private final IEventSerializer eventSerializer;

    public MongoDBEventStore() {
        String host = Proper.get("mongodb.connect.host");
        int prot = Proper.getInt("mongodb.connect.port", 27017);
        Assert.hasText(host, "无法获取MongoDB服务端地址");
        MongoClientOptions.Builder build = new MongoClientOptions.Builder();
        build.connectionsPerHost(Proper.getInt("mongodb.connect.poolsize", 50));
        build.maxWaitTime(Proper.getInt("mongodb.connect.maxwaittime", 60000));
        build.connectTimeout(Proper.getInt("mongodb.connect.connecttimeout", 60000));
        DatabaseName = Proper.get("mongodb.connect.database", "ztrue");
        mongoClient = new MongoClient(new ServerAddress(host, prot), build.build());
        db = mongoClient.getDatabase(DatabaseName);
        serializer = new FastjsonSerializer();
        eventSerializer = new DefaultEventSerializer(serializer, new DefaultClassProvider());
    }

    /**
     * 从领域事件存储中查询指定聚合根事件范围
     *
     * @param aggregateRootId
     * @param aggregateRootTypeName
     * @param minVersion
     * @param maxVersion
     * @return
     */
    @Override
    public List<IDomainEvent> queryAggregateEvents(String aggregateRootId, String aggregateRootTypeName, long minVersion, long maxVersion) throws ClassNotFoundException {
        Assert.hasText(aggregateRootId);
        Assert.hasText(aggregateRootTypeName);
        try {
            MongoCollection<Document> events = db.getCollection(COLLECTIONNAME);
            MongoCursor<Document> eventDocs = events.find(and(eq("aggregateRootId", aggregateRootId), eq("aggregateRootTypeName", aggregateRootTypeName), gte("version", minVersion), lte("version", maxVersion))).iterator();
            List<IDomainEvent> domainEvents = new ArrayList<>();
            Document eventDoc;
            Class eventType;
            while (eventDocs.hasNext()) {
                eventDoc = eventDocs.next();
                domainEvents.addAll(eventSerializer.deserialize(serializer.deserialize(eventDoc.getString("eventdata"), Map.class)));
            }
            return domainEvents;
        } catch (Exception ex) {
            logger.error("保存领域事件时发生异常：" + ex.getMessage(), ex);
            throw ex;
        } finally {

        }
    }

    /**
     * 从领域事件存储中查询指定聚合根
     *
     * @param aggregateRootId
     * @param aggregateRootTypeName
     * @return
     */
    @Override
    public List<IDomainEvent> queryAggregateEvents(String aggregateRootId, String aggregateRootTypeName) throws ClassNotFoundException {
        return queryAggregateEvents(aggregateRootId, aggregateRootTypeName, 0L, Long.MAX_VALUE);
    }

    /**
     * 添加领域事件到存储中
     *
     * @param domainEventStream
     */
    @Override
    public void append(DomainEventStream domainEventStream) {
        try {
            MongoCollection<Document> events = db.getCollection(COLLECTIONNAME);
            Document event = new Document();
            event.put("commandId", domainEventStream.getCommandId());
            event.put("aggregateRootId", domainEventStream.getAggregateRootId());
            event.put("aggregateRootTypeName", domainEventStream.getAggregateRootTypeName());
            event.put("eventdata", serializer.serialize(eventSerializer.serialize(domainEventStream.getDomainEvents())));
            event.put("datetime", new Date());
            event.put("version", domainEventStream.getVersion());
            events.insertOne(event);
        } catch (Exception ex) {
            logger.error("保存领域事件时发生异常：" + ex.getMessage(), ex);
            throw ex;
        } finally {

        }
    }
}
